package com.tang.process;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 对流按键分区后的处理函数的各种定时器demo
 * =====
 * 定时器
 * 1. keyed才有
 * 2. 事件时间定时器，通过watermark来触发的
 * watermark >= 注册的时间
 * 注意：watermark = 当前最大事件时间 - 等待时间  - 1ms, 所以会推迟一条数据
 * 比如，5s的定时器，
 * 如果 等待3s，watermark = 8s - 3s - 1ms = 4999ms，不会触发5s的定时器
 * 需要watermark = 9s - 3s - 1ms = 5999ms，才能触发5s的定时器
 * 3. 在process中获取当前watermark，显示的是上一次的watermark
 *   => 因为process还没接收到这条数据对应生成的新watermark
 *
 * @author tang
 * @since 2023/07/07 17:22
 */
public class KeyedProcessTimerDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        SingleOutputStreamOperator<WaterSensor> sensorDataStream = env.socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(((element, recordTimestamp) -> element.getTs() * 1000L))
                );

        KeyedStream<WaterSensor, String> sensorKeyedStream = sensorDataStream.keyBy(WaterSensor::getId);

        // process：keyed
        sensorKeyedStream.process(new KeyedProcessFunction<String, WaterSensor, String>() {

            /**
             * 来一条数据都会执行一次该方法
             *
             * @param value 输入对象
             * @param ctx 上下文对象
             * @param out 输出对象集合
             */
            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx,
                                       Collector<String> out) {
                // 获取当前数据的key
                String currentKey = ctx.getCurrentKey();

                // 1. 定时器注册
                TimerService timerService = ctx.timerService();
                // 1.1 事件时间的案例
                Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间
                timerService.registerEventTimeTimer(5000L);
                System.out.println("当前key = " + currentKey + ", 当前时间 = " + currentEventTime
                        + ", 注册了一个5秒的定时器");

                // 1.2 处理时间案例
//                long currentProcessingTime = timerService.currentProcessingTime();
//                timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
//                System.out.println("当前key = " + currentKey + ", 当前时间 = " + currentProcessingTime
//                        + ", 注册了一个5秒后的定时器");

                // 1.3 获取process的当前watermark
                long currentWatermark = timerService.currentWatermark();
                System.out.println("当前数据 = " + value + ", 当前watermark = " + currentWatermark);

                // 注册定时器： 处理时间、事件时间
//                timerService.registerProcessingTimeTimer();
//                timerService.registerEventTimeTimer();
                // 删除定时器： 处理时间、事件时间
//                timerService.deleteEventTimeTimer();
//                timerService.deleteProcessingTimeTimer();

                // 获取当前时间进展： 处理时间-当前系统时间，  事件时间-当前watermark
//                 long currentTs = timerService.currentProcessingTime();
//                 long wm = timerService.currentWatermark();

            }

            /**
             * Called when a timer set using {@link TimerService} fires.
             * 时间进展到定时器注册的时间, 调用该方法
             *
             * @param timestamp 当前的触发时间进展
             * @param ctx       上下文
             * @param out       返回值集合，或者叫采集器
             * @throws Exception -
             */
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                String currentKey = ctx.getCurrentKey();
                System.out.println("key = " + currentKey + " 现在的时间是" + timestamp + " 定时器触发");
            }

        });

        env.execute();
    }

}
